package com.hxh.rpc.registry.zookeeper;

import com.hxh.rpc.registry.ServiceRegistry;
import com.hxh.rpc.registry.cache.LocalServiceProviderCache;
import com.hxh.rpc.registry.cache.ServiceProviderCache;
import com.hxh.rpc.registry.model.ServiceMetaConfig;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.*;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.apache.curator.x.discovery.strategies.RoundRobinStrategy;
import org.springframework.beans.factory.annotation.Value;

/**
 * Zookeeper注册中心
 *
 * @author hxh
 * @date 2022/3/2
 * @since 1.0
 */
public class ZookeeperRegistry implements ServiceRegistry {

    private final static String BASE_PATH = "/rpc";
    /**
     * serviceProvider锁
     */
    private final Object lock = new Object();
    /**
     * zk framework client
     */
    private CuratorFramework client;
    /**
     * 服务缓存
     */
    private final ServiceProviderCache cache;
    /**
     * 服务发现
     */
    private ServiceDiscovery<ServiceMetaConfig> serviceDiscovery;

    private ZookeeperProperties zookeeperProperties;

    @Value("${zookeeper.registry.address:localhost:2181}")
    private String address;

    /**
     * @throws Exception
     */
    public ZookeeperRegistry(ZookeeperProperties zookeeperProperties) throws Exception {
        initClient(zookeeperProperties);
        this.client.start();
        this.cache = new LocalServiceProviderCache();
        initServiceDiscovery(zookeeperProperties);
        serviceDiscovery.start();
    }

    private void initServiceDiscovery(ZookeeperProperties zookeeperProperties) {
        JsonInstanceSerializer<ServiceMetaConfig> serializer = new JsonInstanceSerializer<>(ServiceMetaConfig.class);
        serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceMetaConfig.class)
                .client(client)
                .serializer(serializer)
                .basePath(zookeeperProperties.getBasePath())
                .build();
    }

    private void initClient(ZookeeperProperties zookeeperProperties) {
        this.client = CuratorFrameworkFactory.builder().connectString(zookeeperProperties.getAddress())
                .retryPolicy(new ExponentialBackoffRetry(zookeeperProperties.getBaseSleepTimeMs(),
                        zookeeperProperties.getMaxRetries())).build();
    }

    @Override
    public void registry(ServiceMetaConfig serviceMetaConfig) throws Exception {
        ServiceInstance<ServiceMetaConfig> serviceInstance = initServiceInstance(serviceMetaConfig);
        serviceDiscovery.registerService(serviceInstance);
    }

    @Override
    public void unRegistry(ServiceMetaConfig serviceMetaConfig) throws Exception {
        ServiceInstance<ServiceMetaConfig> serviceInstance = initServiceInstance(serviceMetaConfig);
        serviceDiscovery.unregisterService(serviceInstance);
    }

    @Override
    public ServiceMetaConfig discovery(String serviceName) throws Exception {
        //先读缓存
        ServiceProvider<ServiceMetaConfig> serviceProvider = cache.get(serviceName);
        if (serviceProvider == null) {
            synchronized (lock) {
                serviceProvider = serviceDiscovery.serviceProviderBuilder()
                        .serviceName(serviceName)
                        .providerStrategy(new RoundRobinStrategy<>())
                        .build();
                serviceProvider.start();
                this.cache.put(serviceName, serviceProvider);
            }
        }
        ServiceInstance<ServiceMetaConfig> instance = serviceProvider.getInstance();
        return instance != null ? instance.getPayload() : null;
    }

    private ServiceInstance<ServiceMetaConfig> initServiceInstance(ServiceMetaConfig serviceMetaConfig) throws Exception {
        return ServiceInstance.<ServiceMetaConfig>builder()
                .name(serviceMetaConfig.getServiceFullName())
                .address(serviceMetaConfig.getHost())
                .port(serviceMetaConfig.getPort())
                .payload(serviceMetaConfig)
                .uriSpec(new UriSpec("{scheme}://{address}:{port}"))
                .build();
    }
}
